package drds.datasource.connection_pool.impl.thread;

import drds.datasource.connection_pool.impl.DataSource;
import drds.datasource.connection_pool.util.JdbcUtils;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;
import java.sql.SQLException;

@Slf4j
public class CreateConnectionThread extends Thread {


    private final DataSource dataSource;

    public CreateConnectionThread(DataSource dataSource, String name) {
        super(name);
        this.dataSource = dataSource;
        this.setDaemon(true);
    }

    public void run() {
        this.dataSource.iniCountDownLatch.countDown();

        long lastDiscardCount = 0;
        int errorCount = 0;
        for (; ; ) {
            // addLast
            try {
                this.dataSource.lock.lockInterruptibly();
            } catch (InterruptedException e2) {
                break;
            }

            long discardCount = this.dataSource.discardCount;
            boolean discardChanged = discardCount - lastDiscardCount > 0;
            lastDiscardCount = discardCount;

            try {
                boolean emptyWait = true;

                if (this.dataSource.createError != null && this.dataSource.poolingCount == 0
                        && !discardChanged) {
                    emptyWait = false;
                }

                if (emptyWait && this.dataSource.asyncInit
                        && this.dataSource.createCount < this.dataSource.initialSize) {
                    emptyWait = false;
                }

                if (emptyWait) {
                    // 必须存在线程等待，才创建连接
                    if (this.dataSource.poolingCount >= this.dataSource.notEmptyWaitThreadCount
                            && (!(this.dataSource.keepAlive && this.dataSource.activeCount
                            + this.dataSource.poolingCount < this.dataSource.minIdle))
                            && !this.dataSource.isFailContinuous()) {
                        this.dataSource.empty.await();
                    }

                    // 防止创建超过maxActive数量的连接
                    if (this.dataSource.activeCount
                            + this.dataSource.poolingCount >= this.dataSource.maxActive) {
                        this.dataSource.empty.await();
                        continue;
                    }
                }

            } catch (InterruptedException e) {
                this.dataSource.lastCreateError = e;
                this.dataSource.lastErrorTimeMillis = System.currentTimeMillis();

                if (!this.dataSource.closing) {
                    log
                            .error("create connection Thread Interrupted, url: " + this.dataSource.jdbcUrl, e);
                }
                break;
            } finally {
                this.dataSource.lock.unlock();
            }

            Connection connection = null;

            try {
                connection = this.dataSource.createConnection();
            } catch (SQLException e) {
                log.error("create connection SQLException, url: " + this.dataSource.jdbcUrl
                        + ", errorCode " + e.getErrorCode() + ", state " + e.getSQLState(), e);

                errorCount++;
                if (errorCount > this.dataSource.connectionErrorRetryAttempts
                        && this.dataSource.timeBetweenConnectErrorMillis > 0) {
                    // fail over retry attempts
                    this.dataSource.setFailContinuous(true);
                    if (this.dataSource.failFast) {
                        this.dataSource.lock.lock();
                        try {
                            this.dataSource.notEmpty.signalAll();
                        } finally {
                            this.dataSource.lock.unlock();
                        }
                    }

                    if (this.dataSource.breakAfterAcquireFailure) {
                        break;
                    }

                    try {
                        Thread.sleep(this.dataSource.timeBetweenConnectErrorMillis);
                    } catch (InterruptedException interruptEx) {
                        break;
                    }
                }
            } catch (RuntimeException e) {
                log.error("create connection RuntimeException", e);
                this.dataSource.setFailContinuous(true);
                continue;
            } catch (Error e) {
                log.error("create connection Error", e);
                this.dataSource.setFailContinuous(true);
                break;
            }

            if (connection == null) {
                continue;
            }

            boolean result = this.dataSource.put(connection, CreateConnectionTask.create_connection_task_id);
            if (!result) {
                try {
                    JdbcUtils.close(connection);
                } catch (SQLException e) {
                    e.printStackTrace();
                }
                log.info("put physical connection to impl failed.");
            }

            errorCount = 0; // reset errorCount
        }
    }
}